package MicroRpc.framework.protocol.dubbo.jdk;

import MicroRpc.framework.beans.Invoker;
import MicroRpc.framework.commons.StringUtils;
import MicroRpc.framework.beans.Url;
import MicroRpc.framework.excption.CommException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.string.StringDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static MicroRpc.framework.redis.Registry.core.RedisRegistry.isError;
@Slf4j(topic = "m.DubboClient")
public class DubboClient {

    private final Map<String,DubboClientHandler>urlHadnlers=new ConcurrentHashMap<>(16);
    private final Object syncLock=new Object();

    private void dobind(Url url,DubboClientHandler cli) throws InterruptedException {
        EventLoopGroup group;
        Bootstrap bootstrap;
        if (StringUtils.hasText(url.getHost()) && url.getPort()!=null){
            group=new NioEventLoopGroup();
            bootstrap=new Bootstrap();
            bootstrap
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new ProtobufEncoder()); //proto编码器
                            pipeline.addLast(new ByteArrayDecoder());//接收结果直接解码成byte数组
                            pipeline.addLast(cli);
                        }
                    });


            ChannelFuture channelFuture = bootstrap.connect(url.getHost(), url.getPort()).sync();//sync的作用是阻塞,意思是让连接阻塞
            channelFuture.channel().closeFuture().sync();
        }
    }

    public  Object send(Url url, Invoker invoker) throws InterruptedException {
        String urlkey = getUrlkey(url);

        if (!urlHadnlers.containsKey(urlkey)){
            DubboClientHandler clientHandler = new DubboClientHandler();
            urlHadnlers.put(getUrlkey(url),new DubboClientHandler());
            Thread bind = new Thread(() -> {
                try {
                    dobind(url,clientHandler);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            },"bind-thread-"+urlkey);
            bind.start();
            while (true){
                try {
                    clientHandler.setInvoker(invoker);
                    Object result = clientHandler.getResult();
                    if (null==result){
                        throw new CommException("目标调用异常");
                    }
                    urlHadnlers.put(urlkey,clientHandler);
                    return result;
                } catch (NullPointerException e) {
                    Thread.sleep(1500);
                    log.info("第一次与 {} 建立连接,请稍候.",urlkey);
                }
            }
        }
        synchronized (syncLock){
            Object result = null;
            DubboClientHandler dubboClientHandler = urlHadnlers.get(urlkey);
            dubboClientHandler.setInvoker(invoker);
            result = dubboClientHandler.getResult();
            return result;
        }
    }


    private String getUrlkey(Url url){

        return url.getHost()+":"+url.getPort();
    }



}
